import warnings

import numpy as np
import pandas as pd

import plotly.express as px
from plotly.offline import init_notebook_mode
init_notebook_mode(connected=True)

from tqdm import tqdm

warnings.filterwarnings('ignore')

from concurrent.futures import ThreadPoolExecutor

Стратегия и функция полезности

Класс стратегии

Вначале мы задаем вспомогательный класс, представляющий собой открытую позицию, то есть записывает информацию об открытой позиции: размер позиции, цена и булево значение шорт или лонг позиция.

class MyPosition:
    
    def __init__(self, amount: float, price_current: float, short: bool) -> None:
        """
        Инициализация объекта MyPosition.

        :param amount: Количество активa.
        :param price_current: Текущая цена актива.
        :param acc_fees: Количесво накопленных комиссий.
        :param short: Флаг короткой позиции (True, если короткая; False, если длинная).
        """
        self._amount: float = amount
        self._price_current: float = price_current
        self._acc_fees: float = 0
        self._short: bool = short

    def update_state(self, price: float) -> None:
        """
        Обновление состояния позиции.

        :param price: Новая цена актива.
        """
        self._price_current = price
        if self._short:
            # Рассчитываем комиссию за перенос короткой позиции
            transfer_fee = abs(self._amount) * self._price_current * 0.00065
            self._acc_fees += transfer_fee

    def balance(self) -> float:
        """
        Вычисление баланса позиции.

        :return: Баланс позиции.
        """
        return self._amount * self._price_current

Теперь про класс стратегии:

Поля Класса Стратегии

_position: текущая открытая позиция (если есть).

_states: данные о состоянии рынка (временная метка, открытые позиции, цена).

_equity: баланс пользователя.

_margin_equity: сумма использованной маржи.

Основная Функция

Метод run выполняет основную стратегию. Происходит построение скользящей средней, установка диапазона, итерационное отслеживание состояния рынка, а также открытие и закрытие позиций при соответствующих условиях. В коце цикла происходит принудительное закрытие позиции.

Вспомогательные Функции

calc_upper_and_lower: расчет верхней и нижней границ скользящей средней.

open_short: открытие короткой позиции.

open_long: открытие длинной позиции.

close_short: закрытие короткой позиции.

close_long: закрытие длинной позиции.

Параметры

RISK: параметр риска.

STD_COUNT_UP, STD_COUNT_DOWN: коэффициенты для расчета границ скользящей средней.

MA_COUNT: ширина окна для построения скользящей средней.

Данные на выходе

Результат выполнения стратегии записывается в массив, содержащий информацию о времени, цене актива, балансе пользователя, типе позиции, границах скользящей средней и флагах покупки/продажи.

class Strategy:

    params = {'FEE': 0.0004}

    def __init__(self, states, start_equity):
        self._position = None
        self._states = states
        self._equity = start_equity
        self._margin_equity = 0

    def run(self, RISK, STD_COUNT_UP, STD_COUNT_DOWN, MA_COUNT):
        states_ma = self._states['pos'].rolling(window=MA_COUNT).mean()
        data = []

        for i in tqdm(range(len(self._states)), disable=True):
            if i < 10 * MA_COUNT:
                continue
            elif (i == 10 * MA_COUNT):
                state = self._states.loc[i]
                pos_ma = states_ma[i]
                prev_pos_ma = states_ma[i - 1]
                (pos_ma_upper, pos_ma_lower) = self.calc_upper_and_lower(i, states_ma, STD_COUNT_UP, STD_COUNT_DOWN)
            else:
                state = self._states.loc[i]
                pos_ma = states_ma[i]
                prev_pos_ma = states_ma[i - 1]

            b = 0
            s = 0

            if self._position:
                if state['price'] != self._states['price'][i - 1]:
                    self._position.update_state(state['price'])

                #if prev_pos_ma > pos_ma_lower and pos_ma < pos_ma_lower and self._position._short:
                if prev_pos_ma < pos_ma_upper and pos_ma > pos_ma_upper:
                    pos_ma_upper, pos_ma_lower = self.calc_upper_and_lower(i, states_ma, STD_COUNT_UP, STD_COUNT_DOWN)
                    if self._position._short:
                        b = 1
                #if prev_pos_ma < pos_ma_upper and pos_ma > pos_ma_upper and not self._position._short:
                if prev_pos_ma > pos_ma_lower and pos_ma < pos_ma_lower:
                    pos_ma_upper, pos_ma_lower = self.calc_upper_and_lower(i, states_ma, STD_COUNT_UP, STD_COUNT_DOWN)
                    if not self._position._short:
                        s = 1

                if i == (len(self._states) - 2) and self._position._short:
                    b = 1
                if i == (len(self._states) - 2) and not self._position._short:
                    s = 1
                    
                data.append([
                    state['datetime'],
                    state['price'],
                    self._equity - self._position._acc_fees,
                    self._position._short,
                    pos_ma_upper,
                    pos_ma_lower,
                    pos_ma,
                    b,
                    s
                ])

                if b:
                    self.close_short() 
                if s:
                    self.close_long()
            else:
                if prev_pos_ma < pos_ma_upper and pos_ma > pos_ma_upper:
                    pos_ma_upper, pos_ma_lower = self.calc_upper_and_lower(i, states_ma, STD_COUNT_UP, STD_COUNT_DOWN)
                    self.open_long(state, RISK)
                    b = 1
                if prev_pos_ma > pos_ma_lower and pos_ma < pos_ma_lower:
                    pos_ma_upper, pos_ma_lower = self.calc_upper_and_lower(i, states_ma, STD_COUNT_UP, STD_COUNT_DOWN)
                    self.open_short(state, RISK)
                    s = 1

                data.append([
                    state['datetime'],
                    state['price'],
                    self._equity,
                    0,
                    pos_ma_upper,
                    pos_ma_lower,
                    pos_ma,
                    b,
                    s
                ])

        return pd.DataFrame(data, columns=['datetime', 'price', 'equity', 'short', 'pos_ma_upper', 'pos_ma_lower', 'pos_ma', 'buy', 'sell'])

    def calc_upper_and_lower(self, i, states_ma, STD_COUNT_UP, STD_COUNT_DOWN):
        #mean_pos_ma = states_ma.head(i).mean()
        std_pos_ma = states_ma.head(i).std()
        pos_ma_upper = states_ma[i] + STD_COUNT_UP * std_pos_ma
        pos_ma_lower = states_ma[i] - STD_COUNT_DOWN * std_pos_ma
        return pos_ma_upper, pos_ma_lower

    def open_short(self, state, RISK):
        if self._position:
            raise Exception(f'Cannot open position, already have one {self._position}')

        self._margin_equity += (np.floor((1 - self.params['FEE']) * RISK * self._equity / state['price']) * state['price'] - self.params['FEE'] * self._equity)

        amount = (-1) * np.floor((1 - self.params['FEE']) * RISK * self._equity / state['price'])

        self._position = MyPosition(amount, state['price'], True)

    def open_long(self, state, RISK):
        if self._position:
            raise Exception(f'Cannot open position, already have one {self._position}')

        self._margin_equity -= (np.floor((1 - self.params['FEE']) * RISK * self._equity / state['price']) * state['price'] + self.params['FEE'] * RISK * self._equity)

        amount = np.floor((1 - self.params['FEE']) * RISK * self._equity / state['price'])

        self._position = MyPosition(amount, state['price'], False)

    def close_short(self):
        if self._position._short == 0:
            raise Exception(f'Cannot close short position, it is long')

        self._equity += (
                self._margin_equity + (1 - self.params['FEE']) * self._position.balance() - self._position._acc_fees)
        self._margin_equity = 0
        self._position = None

    def close_long(self):
        if self._position._short == 1:
            raise Exception(f'Cannot close long position, it is short')

        self._equity += (self._margin_equity + (1 - self.params['FEE']) * self._position.balance() - self._position._acc_fees)
        self._margin_equity = 0
        self._position = None

Вспомогательные функции

Очистка данных

Отделение необходимых данных из двух csv файлов. А также разделение их на тренировочную и тестовую часть.

def clear_data(ticker):
     # Pos

    df = pd.read_csv(f'{ticker}_full_date.csv', sep=",")
    df = df.sort_values(by=['ticker', 'tradedate', 'tradetime']).drop_duplicates().reset_index().drop('index', axis=1)
    df['datetime'] = pd.to_datetime(df['tradedate'] + ' ' + df['tradetime']) 
    df = df.drop(['tradedate', 'tradetime'], axis=1) 
    df = df[df['clgroup'] == 'YUR'].reset_index().drop('index', axis=1)
    df = df[['datetime', 'pos']]
    data_pos = df

    # Price

    df = pd.read_csv(f'{ticker}_full_date_price.csv', sep=",")
    df.reset_index(inplace=True)
    df.rename(columns={'TRADEDATE': 'datetime'}, inplace=True)
    df['datetime'] = pd.to_datetime(df['datetime'])
    df = df[df['BOARDID']=='TQBR']
    df = df[['datetime', 'WAPRICE']]
    df.rename(columns={'WAPRICE': 'price'}, inplace=True)
    data_price = df

    # All data

    price_dict = dict(zip(data_price['datetime'].dt.date, data_price['price']))

    data_pos['price'] = data_pos['datetime'].dt.date.map(price_dict)

    all_data = data_pos.sort_values('datetime').drop_duplicates().dropna().reset_index().drop('index', axis=1)

    all_data_copy = all_data.copy()

    all_data_copy['date'] = all_data_copy['datetime'].dt.date

    # For train

    train_data = all_data_copy[all_data_copy['date'] < pd.to_datetime('2022-02-20').date()].reset_index().drop('index', axis=1)

    # For test

    test_data = all_data_copy[all_data_copy['date'] > pd.to_datetime('2022-12-31').date()].reset_index().drop('index', axis=1)

    return train_data, test_data

Тренировка

Подбор параметров среди 700 сочетаний параметров с помощью функции полезности. Вначале находится её максимум, далее рассматриваются сочетания, где функция полезности не меньше (max_utility - 0.1). И среди них находится максимум уже по доходности.

def train(all_data_copy1: pd.DataFrame, ticker) -> tuple:
    std_counts_up = np.arange(0.1, 2, 0.2)
    std_counts_down = np.arange(0.1, 2, 0.2)
    ma_counts = np.arange(200, 2200, 300)

    radius = 2

    parameters = []
    final_total_balances = []

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = []

        for std_count_up in std_counts_up:
            for std_count_down in std_counts_down:
                for ma_count in ma_counts:
                    future = executor.submit(run_strategy, all_data_copy1, std_count_up, std_count_down, ma_count)
                    futures.append((std_count_up, std_count_down, ma_count, future))

        for std_count_up, std_count_down, ma_count, future in futures:
            final_total_balance = future.result()
            parameters.append((std_count_up, std_count_down, ma_count))
            final_total_balances.append(final_total_balance)

    data_parameters = pd.DataFrame(parameters, columns=['std_count_up', 'std_count_down', 'ma_count'])
    data_parameters['final_total_balance'] = final_total_balances
    data_parameters['utility'] = 0

    for index_std_count_up in range(len(std_counts_up)):
        for index_std_count_down in range(len(std_counts_down)):
            for index_ma_count in range(len(ma_counts)):
                target_index = index_std_count_up * len(std_counts_down) * len(ma_counts) + index_std_count_down * len(ma_counts) + index_ma_count
                data_parameters.at[target_index, 'utility'] = utility_assess(data_parameters, index_std_count_up, index_std_count_down, index_ma_count, len(std_counts_up), len(std_counts_down), len(ma_counts), radius)

    max_utility = max(data_parameters['utility'])
    max_final_total_balance = max(data_parameters.loc[data_parameters['utility'] > max_utility - 0.1]['final_total_balance'])
    corresponding_parameters = (data_parameters.loc[data_parameters['final_total_balance'] == max_final_total_balance]['std_count_up'].values[0],
                                data_parameters.loc[data_parameters['final_total_balance'] == max_final_total_balance]['std_count_down'].values[0],
                                data_parameters.loc[data_parameters['final_total_balance'] == max_final_total_balance]['ma_count'].values[0])
    corresponding_utility = data_parameters.loc[data_parameters['final_total_balance'] == max_final_total_balance]['utility'].values[0]

    print(f"Ticker: {ticker}")
    print(f"Maximum Utility: {max_utility}")
    print(f"Maximum Total Balance: {max_final_total_balance}")
    print(f"Corresponding Utility: {corresponding_utility}")
    print(f"Corresponding Parameters (std_count, std_timerange, ma_count): {corresponding_parameters}")
    return corresponding_parameters

Тест

def test(all_data_copy2: pd.DataFrame, std_count_up: float, std_count_down: float, ma_count: float) -> float:
    strategy = Strategy(all_data_copy2, 10000000)
    df = pd.DataFrame(strategy.run(1, std_count_up, std_count_down, ma_count))
    df['equity'] = df['equity'] / df['equity'].iloc[0]
    final_total_balance = df['equity'].iloc[-1]

    print(f'Test profit = {final_total_balance}')
    return final_total_balance

Для многопоточности

Функции для примения многопоточности в программе.

def process_ticker(ticker):
    train_data, test_data = clear_data(ticker)
    corresponding_parameters = train(train_data, ticker)
    test(test_data, *corresponding_parameters)

def run_strategy(all_data_copy, std_count_up, std_count_down, ma_count):
    strategy = Strategy(all_data_copy, 10000000)
    df = pd.DataFrame(strategy.run(1, std_count_up, std_count_down, ma_count))
    df['equity'] = df['equity'] / df['equity'].iloc[0]
    final_total_balance = df['equity'].iloc[-1]
    return final_total_balance

Функция полезности

Тут мы задаём некоторую функцию полезности и затем считаем ее для каждого параметра, в качестве радиуса мы берём 2, то есть по рассматриваем дисперсию окресности из 125 $((2*2+1)^3)$ точек.

def utility_fun(income, var):
    return income - 10 * var

def utility_assess(df_plotly, index_std_count_up, index_std_count_down, index_ma_count, len_std_counts_up, len_std_counts_down, len_ma_counts, radius):
    income = df_plotly.loc[
        (df_plotly['std_count_up'] == df_plotly['std_count_up'].iloc[index_std_count_up]) &
        (df_plotly['std_count_down'] == df_plotly['std_count_down'].iloc[index_std_count_down]) &
        (df_plotly['ma_count'] == df_plotly['ma_count'].iloc[index_ma_count])
    ]['final_total_balance'].values[0]

    index_std_count_up -= radius
    index_std_count_down -= radius
    index_ma_count -= radius

    array_income = np.empty((2 * radius + 1, 2 * radius + 1, 2 * radius + 1))

    for i in range(2 * radius + 1):
        for j in range(2 * radius + 1):
            for k in range(2 * radius + 1):
                index_i = index_std_count_up + i
                index_j = index_std_count_down + j
                index_k = index_ma_count + k

                if (
                    0 <= index_i < len_std_counts_up and
                    0 <= index_j < len_std_counts_down and
                    0 <= index_k < len_ma_counts
                ):
                    array_income[i, j, k] = df_plotly.loc[
                        (df_plotly['std_count_up'] == df_plotly['std_count_up'].iloc[index_i]) &
                        (df_plotly['std_count_down'] == df_plotly['std_count_down'].iloc[index_j]) &
                        (df_plotly['ma_count'] == df_plotly['ma_count'].iloc[index_k])
                    ]['final_total_balance'].values[0]

    std_of_income = np.std(array_income[~np.isnan(array_income)]) if np.sum(~np.isnan(array_income)) >= 2 else 0
    utility = utility_fun(income, std_of_income)
    return utility

Применение

Подбор оптимальных параметров и вывод доходности.

ticker_names=['sr', 'gz', 'lk', 'vb', 'rn', 'mn', 'af', 'al', 'sn', 'yn', 'tt', 'nm', 'hy', 'me', 'fv', 'gk', 'mg']

with ThreadPoolExecutor() as executor:
    executor.map(process_ticker, ticker_names)

Ticker: sn
Maximum Utility: 0.31788352359963856
Maximum Total Balance: 1.3445981059217469
Corresponding Utility: 0.25376568222276386
Corresponding Parameters (std_count, std_timerange, ma_count): (1.1000000000000003, 0.7000000000000001, 200)
Test profit = 1.3296202276732652
Ticker: tt
Maximum Utility: 0.7413407906539069
Maximum Total Balance: 1.0
Corresponding Utility: -0.24055437492245002
Corresponding Parameters (std_count, std_timerange, ma_count): (0.7000000000000001, 1.1000000000000003, 2000)
Test profit = 1.150119740559525
Ticker: mn
Maximum Utility: 0.7707334583417212
Maximum Total Balance: 1.0
Corresponding Utility: -2251.9078693800307
Corresponding Parameters (std_count, std_timerange, ma_count): (0.1, 1.9000000000000004, 2000)
Ticker: rn
Maximum Utility: 0.6261486753719142
Maximum Total Balance: 0.996677589936
Corresponding Utility: 0.5424869624285997
Corresponding Parameters (std_count, std_timerange, ma_count): (0.5000000000000001, 1.9000000000000004, 2000)
Test profit = 1.1509129037672001
Test profit = 0.87408150129375
Ticker: af
Maximum Utility: 0.32667065666462475
Maximum Total Balance: 0.9164965757988631
Corresponding Utility: 0.24412374914941215
Corresponding Parameters (std_count, std_timerange, ma_count): (1.5000000000000004, 1.3000000000000003, 1100)
Test profit = 1.340699591308288
Ticker: nm
Maximum Utility: 0.790710850315741
Maximum Total Balance: 1.0861413002256
Corresponding Utility: 0.762506749378346
Corresponding Parameters (std_count, std_timerange, ma_count): (1.5000000000000004, 0.1, 2000)
Test profit = 0.982291381394388
Ticker: al
Maximum Utility: 0.6961923382382422
Maximum Total Balance: 1.0301410478167294
Corresponding Utility: 0.6961923382382422
Corresponding Parameters (std_count, std_timerange, ma_count): (1.5000000000000004, 0.5000000000000001, 2000)
Test profit = 0.7632387064492832
Ticker: yn
Maximum Utility: 0.5645708943478455
Maximum Total Balance: 0.9158679463687522
Corresponding Utility: 0.5232823541471745
Corresponding Parameters (std_count, std_timerange, ma_count): (1.1000000000000003, 1.9000000000000004, 200)
Test profit = 1.1444019765163957
Ticker: lk
Maximum Utility: 0.44546966860032255
Maximum Total Balance: 1.018224037645488
Corresponding Utility: 0.44546966860032255
Corresponding Parameters (std_count, std_timerange, ma_count): (1.3000000000000003, 1.9000000000000004, 2000)
Test profit = 0.6038929560625
Ticker: sr
Maximum Utility: 0.5607966519020064
Maximum Total Balance: 1.1022781941188184
Corresponding Utility: 0.5463749703404512
Corresponding Parameters (std_count, std_timerange, ma_count): (0.9000000000000001, 0.5000000000000001, 1400)
Test profit = 1.0282202039989643
Ticker: gz
Maximum Utility: 0.1843111111429132
Maximum Total Balance: 0.7386922179780796
Corresponding Utility: 0.15246636744707198
Corresponding Parameters (std_count, std_timerange, ma_count): (0.5000000000000001, 0.9000000000000001, 200)
Test profit = 1.137889739611637
Ticker: vb
Maximum Utility: 1.1560153379556783
Maximum Total Balance: 0.7789406628697979
Corresponding Utility: 1.1560153379556783
Corresponding Parameters (std_count, std_timerange, ma_count): (0.5000000000000001, 1.9000000000000004, 2000)
Test profit = 1.0172080985527603
Ticker: gk
Maximum Utility: 0.8073026145896098
Maximum Total Balance: 1.02242012544
Corresponding Utility: 0.026212750664499462
Corresponding Parameters (std_count, std_timerange, ma_count): (0.9000000000000001, 0.30000000000000004, 2000)
Test profit = 0.8533278441854918
Ticker: hy
Maximum Utility: 0.7909200273841779
Maximum Total Balance: 1.0444722672864002
Corresponding Utility: 0.6018266950103128
Corresponding Parameters (std_count, std_timerange, ma_count): (1.3000000000000003, 0.1, 2000)
Ticker: me
Maximum Utility: 0.797714891742952
Maximum Total Balance: 1.2169686430581383
Corresponding Utility: 0.797714891742952
Corresponding Parameters (std_count, std_timerange, ma_count): (1.5000000000000004, 0.5000000000000001, 800)
Test profit = 1.0997172356807976
Test profit = 0.5230073084245509
Ticker: mg
Maximum Utility: 0.5389506079936247
Maximum Total Balance: 1.140050052375
Corresponding Utility: 0.511356770226375
Corresponding Parameters (std_count, std_timerange, ma_count): (1.5000000000000004, 0.1, 1700)
Test profit = 0.5830371836850659
Ticker: fv
Maximum Utility: 0.5785285242951834
Maximum Total Balance: 0.975891574840926
Corresponding Utility: 0.5785285242951834
Corresponding Parameters (std_count, std_timerange, ma_count): (0.9000000000000001, 0.9000000000000001, 200)
Test profit = 0.9166081184728172

Средняя доходность составлвляет 0,9704867481

Отладочная печать

Пример работы стратегии на оптимальных параметрах для акций ПАО "Сургутнефтегаз".

Первый график показывает построение диапазона для траектории количества позиций.

Второй - баланс пользователя.

На третьем изображены точки открытия/закрытия позиций. Если сначала зелёная, а затем красная, то это лонг. Если наоборот, то шорт.

train_data, test_data = clear_data('sn')

strategy = Strategy(test_data, 10000000)
df = pd.DataFrame(strategy.run(1, 1.1, 0.7, 200))

px.line(df, x='datetime', y=['pos_ma', 'pos_ma_upper', 'pos_ma_lower']).show()

px.line(df, x='datetime', y='equity').update_xaxes(type='category').show()

up = pd.DataFrame(columns=['datetime', 'price'])
down = pd.DataFrame(columns=['datetime', 'price'])
for i in range(len(df['price'])):
    if df['buy'][i]:
        up.loc[len(up)] = df.iloc[i]
    elif df['sell'][i]:
        down.loc[len(down)] = df.iloc[i]
fig = px.line(df, x='datetime', y='price', title='График цен по времени', labels={'datetime': 'Дата и время', 'price': 'Цена'})
fig.add_trace(px.scatter(up, x='datetime', y='price').update_traces(marker=dict(color='green')).data[0])
fig.add_trace(px.scatter(down, x='datetime', y='price').update_traces(marker=dict(color='red')).data[0])
fig.update_xaxes(type='category')
fig.show()

Визуализация функции полезности

import plotly.graph_objs as go
from plotly.subplots import make_subplots

train_data, test_data = clear_data('sn')

step_std_counts_up = 0.2
step_std_counts_down = 0.2
step_ma_counts = 300
std_counts_up = np.arange(0.1, 2, step_std_counts_up)
std_counts_down = np.arange(0.1, 2, step_std_counts_down)
ma_counts = np.arange(200, 2200, step_ma_counts)

parameters = []
final_total_balances = []

with ThreadPoolExecutor(max_workers=4) as executor:
    futures = []
    
    for std_count_up in std_counts_up:
        for std_count_down in std_counts_down:
            for ma_count in ma_counts:
                future = executor.submit(run_strategy, train_data, std_count_up, std_count_down, ma_count)
                futures.append((std_count_up, std_count_down, ma_count, future))

    for std_count_up, std_count_down, ma_count, future in futures:
        final_total_balance = future.result()
        parameters.append((std_count_up, std_count_down, ma_count))
        final_total_balances.append(final_total_balance)

# Преобразование в DataFrame для удобства работы с данными
df_plotly = pd.DataFrame(parameters, columns=['std_count_up', 'std_count_down', 'ma_count'])
df_plotly['final_total_balance'] = final_total_balances

# Создание трехмерного графика в Plotly
fig = make_subplots(rows=1, cols=1, specs=[[{'type': 'scatter3d'}]])

scatter = go.Scatter3d(
    x=df_plotly['std_count_up'],
    y=df_plotly['std_count_down'],
    z=df_plotly['ma_count'],
    mode='markers',
    marker=dict(
        size=4,
        color=df_plotly['final_total_balance'],
        colorscale='Viridis',
        colorbar=dict(title='Final Total Balance')
    )
)

fig.add_trace(scatter)

# Наименование осей
fig.update_layout(scene=dict(
                    xaxis_title='std_count_up',
                    yaxis_title='std_count_down',
                    zaxis_title='ma_count')
                 )

# Отображение графика
fig.show()

Видно, что максимальные значения находяться в (1.1, 0.1, 200), но соседние точки имеют намного меньше доходности. А значит она нам не подходит.

radius = 2

array_utility = []
for index_std_count_up in range(len(std_counts_up)):
    for index_std_count_down in range(len(std_counts_down)):
        for index_ma_count in range(len(ma_counts)):
            array_utility.append(utility_assess(df_plotly, index_std_count_up, index_std_count_down, index_ma_count, len(std_counts_up), len(std_counts_down), len(ma_counts), radius))

df_plotly['final_utility_balance'] = array_utility
print(np.max(df_plotly['final_total_balance']), np.max(array_utility))

1.6428116300105169 0.31788352359963856

import plotly.graph_objs as go
from plotly.subplots import make_subplots

# Создание трехмерного графика в Plotly
fig = make_subplots(rows=1, cols=1, specs=[[{'type': 'scatter3d'}]])

scatter = go.Scatter3d(
    x=df_plotly['std_count_up'],
    y=df_plotly['std_count_down'],
    z=df_plotly['ma_count'],
    mode='markers',
    marker=dict(
        size=4,
        color=df_plotly['final_utility_balance'],
        colorscale='Viridis',
        colorbar=dict(title='Final Total Balance')
    ),
    text=df_plotly.apply(lambda row: f' final_total_balance: {row["final_total_balance"]}final_utility_balance: {row["final_utility_balance"]}', axis=1)
)

fig.add_trace(scatter)

# Наименование осей
fig.update_layout(scene=dict(
                    xaxis_title='std_count_up',
                    yaxis_title='std_count_down',
                    zaxis_title='ma_count')
                 )

# Отображение графика
fig.show()

На этом графике видно, что около появляется область точек, где функция полезности имеет около максимальные значения. Это используется при подборе оптимальных параметров.